/****************************************************************************
 * Copyright 2010 Huascar A. Sanchez                                        *
 *                                                                          *
 * Licensed under the Apache License, Version 2.0 (the "License");          *
 * you may not use this file except in compliance with the License.         *
 * You may obtain a copy of the License at                                  *
 *                                                                          *
 *     http://www.apache.org/licenses/LICENSE-2.0                           *
 *                                                                          *
 * Unless required by applicable law or agreed to in writing, software      *
 * distributed under the License is distributed on an "AS IS" BASIS,        *
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. *
 * See the License for the specific language governing permissions and      *
 * limitations under the License.                                           *
 ****************************************************************************/
package com.gotobject.greatq.datastore;

import org.jiql.db.ColumnInfo;
import org.jiql.db.Row;
import org.jiql.db.TableInfo;
import org.jiql.db.Union;
import org.jiql.db.jdbc.stat.StatementProcessor;
import org.jiql.db.jiqlDBMgr;
import org.jiql.db.objs.jiqlCellValue;
import org.jiql.db.objs.jiqlTableInfo;
import org.jiql.jdbc.ResultSet;
import org.jiql.jdbc.Statement;
import org.jiql.util.Criteria;
import org.jiql.util.Gateway;
import org.jiql.util.JGException;
import org.jiql.util.SQLParser;
import tools.util.NameValue;

import java.lang.reflect.Field;
import java.sql.SQLException;
import java.util.Deque;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.List;
import java.util.Vector;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import static com.gotobject.filter.internal.Objects.cast;

/**
 * @author huascar.sanchez@gmail.com (Huascar A. Sanchez)
 */
public class DatastoreStatement extends Statement {
    private final CopyOnWriteArrayList<ResultSet> allResults;
    private final Deque<String> internalBatch;
    private final AtomicInteger index;
    private final AtomicReference<DatastoreSQLParser> atomicParser = new AtomicReference<DatastoreSQLParser>();


    public DatastoreStatement(DatastoreConnection connection) throws SQLException {
        super(connection);
        index           = new AtomicInteger(0);
        allResults      = new CopyOnWriteArrayList<ResultSet>();
        internalBatch   = new LinkedBlockingDeque<String>();
    }
    public DatastoreStatement(Statement statement) throws SQLException {
        this((DatastoreConnection) statement.getConnection());
    }


    @Override
    public void addBatch(String sql) throws SQLException {
        if(!internalBatch.contains(sql)) internalBatch.add(sql);
    }

    @Override
    public int[] executeBatch() throws SQLException {
        final Feedback feedback  = new Feedback();
        while(internalBatch.size() > 0){
            execute(internalBatch.poll(), feedback);
            try {
                allResults.addIfAbsent(feedback.call());
                feedback.clear();
            } catch (Exception e) {
                throw new SQLException(e);
            }
        }

        return new int[0];
    }

    @Override
    public SQLParser getSQLParser() {
        return atomicParser.get();
    }

    void execute(String sql, Feedback feedback) throws SQLException {
        execute(sql);
        feedback.set(super.getResultSet());
    }

    @Override
    public boolean execute(String sql) throws SQLException {
        final DatastoreConnection activeConnection = cast(getConnection(),  DatastoreConnection.class);
        // todo(Huascar) pay attention to any side effects -> locals static variable was removed
        @SuppressWarnings({"RedundantTypeArguments"})
        final Vector<String> activeLocals  = this.<Vector<String>>getField(super.getClass().getSuperclass(), "locals", true);
        final boolean isMaintainanceQuery  = ( activeLocals.contains(sql) || sql.toLowerCase().trim().startsWith("load ") || sql.toLowerCase().trim().startsWith("insert "));
        @SuppressWarnings({"RedundantTypeArguments"})
        final Hashtable activeDirectValues = this.<Hashtable>getField(
                super.getClass().getSuperclass(),
                "directValues",
                false
        );
        if(activeConnection.isRemote() && !isMaintainanceQuery){
            @SuppressWarnings({"unchecked"})
            final Hashtable source = new DataFeeder(
                    activeConnection,
                    sql,
                    activeDirectValues
            ).pullData();



            final Object remoteError = source.get("error");

            if (remoteError != null){
			    throw new SQLException(remoteError.toString());
            }

		    final DatastoreSQLParser sqp = new DatastoreSQLParser((SQLParser) source.get("sqlparser"));//(DatastoreSQLParser)source.get("sqlparser");
            atomicParser.compareAndSet(atomicParser.get(), sqp);
            final Vector sol =  (Vector)source.get("results");
            sqp.setConnection(activeConnection);
            if(sol != null){
                setField(super.getClass().getSuperclass(), "resultset", new ResultSet(sol, sqp));
            }

            if(!activeLocals.contains(sqp.getAction())){
                return true;
            }
        }

        atomicParser.compareAndSet(atomicParser.get(), new DatastoreSQLParser(sql, activeConnection));
        final DatastoreSQLParser sqp = atomicParser.get();
        if(sqp.getAction() != null){
            final Gateway   gappe = Gateway.get(activeConnection.getProperties());
            final Union     union = sqp.getUnion();
            sqp.mergeAliases(union.getAliases());
            final StatementProcessor sp = sqp.getStatementProcessor();
            if(sp != null){
                setField(super.getClass().getSuperclass(), "resultset", sp.process(sqp));
                return true;
            } else if (sqp.getAction().equalsIgnoreCase("createIndex")) {
            } else if (sqp.getAction().equals("setOperation")) {
            } else if (sqp.getAction().equals("loadTable")) {
                sqp.getLoadTable().execute(activeConnection);
            } else if (sqp.getAction().equals("sqlInsert")){
                sqp.getInsertIntoTable().execute(activeConnection,activeDirectValues);
            } else if (sqp.getAction().equals("showTables")) {
                showTables(sqp);
            } else if (sqp.getAction().equals("describeTable")) {
                describeTable(sqp);
            } else if (sqp.getAction().equals("getColumns")) {
                getColumns(sqp);
            } else if (sqp.getAction().equals("getPrimaryKeys")) {
                getPrimaryKeys(sqp);
            } else if (sqp.getAction().equals("getExportedKeys")) {
                getExportedKeys(sqp);
            } else if (sqp.getAction().equals("getIndex")) {
                getIndex(sqp);
            } else if (sqp.getAction().equals("getImportedKeys")) {
                getImportedKeys(sqp);
            } else if (sqp.getAction().equals("getIdentity")) {
                getIdentity(sqp);
            } else if (sqp.getAction().equals("getTypeInfo")){
                getTypeInfo(sqp);
            } else if (sqp.getAction().equals("getFoundRows")) {
                getFoundRows(sqp);
            } else if (sqp.getAction().equals("jiqldescribeTable")) {
                jiqldescribeTable(sqp);
            } else if (sqp.getAction().equals("createTable")) {
                if(Gateway.get(activeConnection.getProperties()).readTableInfo(sqp.getTable()).size() > 0) {
                    if (sqp.getCreateParser().ifNotExists()) return true;
                    throw JGException.get("table_exists",sqp.getTable() + " Table Exists");
                }

                jiqlTableInfo ti = sqp.getJiqlTableInfo(true);
                ti.setPrefix(sqp.hasPrefix());
                ti.setPrefixName(sqp.getPrefixValue());
                ti.setTableLeafs(sqp.hasTableleafs());

                final Hashtable       hash = sqp.getHash();
                final Hashtable       dv   = sqp.getDefaultValues();
                final Enumeration     dven = dv.keys();

                String n;
                String v;
                int ml;
                int ty;
                String dVal;
                while(dven.hasMoreElements()) {
                    n = dven.nextElement().toString();
                    v = (String)hash.get(n);
                    if(v == null) throw JGException.get("no_column_for_default_value","No Column for default value " + n);
                    ty = ColumnInfo.getTypeFromName(v);
                    dVal = (String)dv.get(n);
                    if (!dVal.equalsIgnoreCase("null")) {
                        jiqlCellValue.getObj(dVal,ty,sqp);
                    }

                    if (v.startsWith("varchar(")){
                        v = v.substring("varchar(".length(),v.length()-1);
                        ml =  Integer.parseInt(v);

                        if (!dVal.equalsIgnoreCase("null"))  {
                            if (dVal.length() > ml) {
                                throw JGException.get("default_value_cannot_exceed_max_column_length",v + " Default value cannot exceed max column length " + n);
                            }
                        }

                    }
                }

                gappe.writeTableInfo(sqp.getTable(),hash);
                jiqlDBMgr.get(sqp.getProperties()).getCommand("addPrimaryKeys").execute(sqp);
                ti.setDefaultValues(sqp.getDefaultValues());
                ti.setNotNulls(sqp.getNotNulls());
                jiqlDBMgr.get(sqp.getProperties()).saveTableInfo(sqp.getTable(),ti);

            } else if (sqp.getAction().equals("dropTable")) {
                jiqlDBMgr.get(sqp.getProperties()).getCommand("dropTable").execute(sqp);
		    } else if (sqp.getAction().equals("writeTableRow")) {
                jiqlDBMgr.get(sqp.getProperties()).getCommand("VerifyTable").execute(sqp);


                @SuppressWarnings({"unchecked"})
                final Hashtable<String, Object> wt = sqp.getHash();
                if (activeDirectValues.size() > 0){
                    final Enumeration en = wt.keys();
                    String k;
                    String v;
                    Object dv;

                    while (en.hasMoreElements()){
                        k = en.nextElement().toString();
                        v = wt.get(k).toString();
                        dv = activeDirectValues.get(v);

                        if (dv != null)
                            wt.put(k,dv);
                        }
                }

                activeDirectValues.clear();

			    jiqlDBMgr.get(sqp.getProperties()).getCommand("verifyPrimaryKeys").execute(sqp);
			    jiqlDBMgr.get(sqp.getProperties()).getCommand("verifyConstraints").execute(sqp);
			    jiqlDBMgr.get(sqp.getProperties()).getCommand("VerifyDefaultValues").execute(sqp);
			    jiqlDBMgr.get(sqp.getProperties()).getCommand("VerifyNotNulls").execute(sqp);
			    jiqlDBMgr.get(sqp.getProperties()).getCommand("VerifyTypeValues").execute(sqp);

			    Gateway.get(activeConnection.getProperties()).writeTableRow(sqp.getTable(),wt,sqp);
				activeConnection.setIdentity(sqp.getInsertParser().getAutoIncrementValue());
    	    } else if (sqp.getAction().equals("select")) {
                select(sqp);
		    } else if (sqp.getAction().equals("addConstraint")) {
			    jiqlDBMgr.get(sqp.getProperties()).getCommand("addConstraint").execute(sqp);
		    } else if (sqp.getAction().equals("update")) {
                Hashtable h = Gateway.get(activeConnection.getProperties()).readTableValue(
                        sqp.getTable(),
                        sqp.getIncludeAllList(),
                        sqp.getSelectList(),
                        sqp.getEitherOrAllList(),
                        sqp.isDistinct(),
                        sqp
                );

                if (h == null) h = new Hashtable();
		        Enumeration en = h.keys();
		        String rid;
		        Vector ul = sqp.getUpdateList();
		        NameValue nv;
		        Row ur;
		        Integer updatedRowsCount = 0;
		        while (en.hasMoreElements()) {
			        rid = en.nextElement().toString();
			        ur = (Row)h.get(rid);
			        for (int ct = 0; ct < ul.size();ct++) {
                        nv = (NameValue)ul.elementAt(ct);
			            ur.put(nv.name,sqp.convert(nv.value,nv.name));
			        }

                    gappe.updateTableValue( sqp.getTable(),ur,sqp);
                    updatedRowsCount ++;
		        }

                if(updatedRowsCount > 0){
                    setField(super.getClass().getSuperclass(), "updatedRowsCount", updatedRowsCount);
                }
            } else if (sqp.getAction().equals("delete")) {
                TableInfo tp = Gateway.get(activeConnection.getProperties()).readTableInfo(sqp.getTable());
                if (tp == null || tp.size() < 1)
					 throw JGException.get("table_not_exist","Table does not exists! " + sqp.getTable());

                Vector vil = sqp.getIncludeAllList() ;
                for (int ctvil = 0; ctvil < vil.size(); ctvil++) {
                    final Criteria cr = (Criteria)vil.elementAt(ctvil);
                    if (tp.getColumnInfo(cr.getName()) == null)
                        throw JGException.get("column_not_exist","Column does not exists! " + cr.getName());
                }

                Hashtable h = Gateway.get(activeConnection.getProperties()).readTableValue(
                        sqp.getTable(),
                        sqp.getIncludeAllList(),
                        sqp.getSelectList(),
                        sqp.getEitherOrAllList(),
                        sqp.isDistinct(),
                        sqp
                );

                if (h == null) return true;
                final Enumeration en = h.keys();
                Integer updatedRowsCount = 0;
                while (en.hasMoreElements()) {
                    final String rid = en.nextElement().toString();
                    gappe.deleteTableValue(sqp.getTable(),rid);
                    updatedRowsCount ++;
                }

                if(updatedRowsCount > 0){
                    //noinspection UnnecessaryBoxing
                    setField(super.getClass().getSuperclass(), "updatedRowsCount", updatedRowsCount);
                }

            } else {
                throw JGException.get(
                        "statement_not_recognized",
                        sqp.getAction() + " Statement NOT recognized! " + sql
                );
            }
        }

        return true;
   }


    @Override
    public ResultSet executeQuery(String sql) throws SQLException {
        final DatastoreConnection activeConnection = cast(getConnection(),  DatastoreConnection.class);
        // todo(Huascar) pay attention to any side effects -> locals static variable was removed
        @SuppressWarnings({"RedundantTypeArguments"})
        final Vector<String> activeLocals  = this.<Vector<String>>getField(super.getClass().getSuperclass(), "locals", true);
        final boolean isMaintainanceQuery  = (activeLocals.contains(sql));
        @SuppressWarnings({"RedundantTypeArguments"})
        final Hashtable activeDirectValues = this.<Hashtable>getField(
                super.getClass().getSuperclass(),
                "directValues",
                false
        );

        if(activeConnection.isRemote() && !isMaintainanceQuery){
            @SuppressWarnings({"unchecked"})
            final Hashtable source = new DataFeeder(
                    activeConnection,
                    sql,
                    activeDirectValues
            ).pullData();

            final Object remoteError = source.get("error");

            if (remoteError != null){
			    throw new SQLException(remoteError.toString());
            }

		    final DatastoreSQLParser sqp = new DatastoreSQLParser((SQLParser) source.get("sqlparser"));//(DatastoreSQLParser)source.get("sqlparser");
            atomicParser.compareAndSet(atomicParser.get(), sqp);
            final Vector sol =  (Vector)source.get("results");
            sqp.setConnection(activeConnection);
            if(sol != null){
                allResults.addIfAbsent(new ResultSet(sol, sqp));
            }

            if(!activeLocals.contains(sqp.getAction())){
                return getResultSet();
            }
        }

        atomicParser.compareAndSet(atomicParser.get(), new DatastoreSQLParser(sql, activeConnection));
        final DatastoreSQLParser sqp = atomicParser.get();
        if(sqp.getAction() != null){
		    final StatementProcessor sp = sqp.getStatementProcessor();
            if (sp != null){
                allResults.addIfAbsent(sp.process(sqp));
                //setField(super.getClass().getSuperclass(), "resultset", sp.process(sqp));
                return getResultSet();
            } else if (sqp.getAction().equals("select")){
                select(sqp);
            } else if (sqp.getAction().equals("showTables")){
                showTables(sqp);
		    } else if (sqp.getAction().equals("describeTable")){
                describeTable(sqp);
		    } else if (sqp.getAction().equals("getColumns")){
                getColumns(sqp);
		    } else if (sqp.getAction().equals("getPrimaryKeys")){
                getPrimaryKeys(sqp);
		    } else if (sqp.getAction().equals("getIndex")){
                getIndex(sqp);
		    } else if (sqp.getAction().equals("getExportedKeys")){
                getExportedKeys(sqp);
		    } else if (sqp.getAction().equals("getTypeInfo")){
                getTypeInfo(sqp);
		    } else if (sqp.getAction().equals("getIdentity")){
                getIdentity(sqp);
		    } else if (sqp.getAction().equals("getFoundRows")){
                getFoundRows(sqp);
		    } else if (sqp.getAction().equals("getImportedKeys")){
                getImportedKeys(sqp);
		    } else if (sqp.getAction().equals("jiqldescribeTable")){
                jiqldescribeTable(sqp);
		    }
        }
        return getResultSet();  
    }

    private void setField(Class<?> klass, String name, Object value) throws SQLException {
        try {
            for(Field each : klass.getDeclaredFields()){
                if(each.getName().equalsIgnoreCase(name)){
                    each.setAccessible(true);
                    each.set(this, value);
                }
            }
        } catch (Exception e){
            throw new SQLException(e);
        }
    }


    private <T> T getField(Class<?> klass, String name, boolean isStatic) throws SQLException {
        try {
            for(Field each : klass.getDeclaredFields()){
                if(each.getName().equalsIgnoreCase(name)){
                    each.setAccessible(true);
                    //noinspection unchecked
                    return (T) each.get(isStatic ? null : this);
                }
            }

            throw new NoSuchFieldException("error: element not found");
        } catch (Exception e) {
            throw new SQLException(e);
        }
    }


    @Override
    public ResultSet getResultSet() throws SQLException {
        return allResults.isEmpty() ? null : allResults.get(index.get());
    }


    boolean containsRecord(int idx) {
        try {
          return allResults.get(idx) != null;  
        } catch (Exception e){
            return false;
        }
    }

    @Override
    public boolean getMoreResults() throws SQLException {
        return index.compareAndSet(index.get(), index.incrementAndGet());
    }

    @Override
    public boolean getMoreResults(int i) throws SQLException {
        throw new UnsupportedOperationException("not supported in this version");
    }

    public List<ResultSet> getAllResults(){
        return allResults;
    }

    /**
     * 
     */
    private static class Feedback implements Callable<ResultSet> {
        private final AtomicReference<ResultSet> current;

        Feedback(){
            current = new AtomicReference<ResultSet>();
        }

        void set(ResultSet newOne){
            current.compareAndSet(current.get(), newOne);
        }

        void clear(){
            set(null);
        }

        @Override
        public ResultSet call() throws Exception {
            return current.get();
        }

    }


}
